#include "comet_subscribe_request.h"

#include <sstream>
#include <iomanip>

#include <boost/lexical_cast.hpp>
#include <boost/bind.hpp>

#include "base/logging.h"
#include "base/json_escape.h"

#include "temp_id_generator.h"
#include "ImServiceCommon_types.h"
#include "subscriber_cache.h"
#include "uid_session_map.h"

#include "im/adapters/session_adapter.h"
#include "im/adapters/user_profile_adapter.h"
#include "rpc_thread_manager.h"
#include "update_session_task.h"
#include "check_ticket_task.h"
#include "notify_presence_task.h"
#include "link_temp_id_qid_task.h"
#include "linger_session_map.h"

using apache::thrift::concurrency::Runnable;
using boost::asio::deadline_timer;

namespace imserver {

extern boost::asio::io_service g_io_service;

int64_t CometSubscribeRequest::ctor_count_ = 0;
int64_t CometSubscribeRequest::out_messages_ = 0;

void CometSubscribeRequest::OnReadError() {
  LOG_INFO("CometSubscribeRequest read error : " << effective_uid() << " " << effective_session());
  // TODO : 读出错的时候，是linger还是?
  AddLingerSession();
}

int CometSubscribeRequest::OnCheckTicketComplete(int64_t qid, const std::string & username) {
  if (qid >= 0) {
    qid_ = qid;
  }
  if (!username.empty()) {
    username_ = username;
  }


  if (qid_ > 0) {
    if (qid_ticket_.empty()) {
      qid_ticket_ = CalculateSession(qid_, cookie("Q"), cookie("T"));
      LOG_INFO("user " << qid_ << " login, session=" << qid_ticket_);
      UserLoginResponse(false);
    } else {
      std::string new_ticket = CalculateSession(qid_, cookie("Q"), cookie("T"));
      LOG_INFO("user " << qid_ << " access move, " << qid_ticket_ << "=>" << new_ticket);
      qid_ticket_ = new_ticket;
      UserLoginResponse(true);
    }
    return 0;
  }

  if (temp_id_ > 0) {
    std::string new_ticket = CalculateSession(temp_id_, "", "");
    LOG_INFO("user " << temp_id_ << " access move, " << temp_id_ticket_ << "=>" << new_ticket);
    temp_id_ticket_ = new_ticket;
    UserLoginResponse(true);
  } else {
    temp_id_ = TempIdGenerator::Instance().Generate();
    temp_id_ticket_ = CalculateSession(temp_id_, "", "");
    LOG_INFO("user " << temp_id_ << " login, session=" << temp_id_ticket_);
    UserLoginResponse(false);
  }
  
  return 0;
}

int CometSubscribeRequest::UserLoginResponse(bool is_access_switch) {
  // TODO : 分配登录或临时身份
  std::stringstream rsp_body;

  if (username_.empty()) {
    username_ = GetDefaultUsername(effective_uid());
  }

  std::string callback = ParseCallback();
  if (!callback.empty()) {
    // 开头空格是 IE bug fix
    rsp_body << ' ' << callback << "(";
  }

  rsp_body << "{type:" << (is_access_switch ? ACCESS_SWITCH : LOGIN_SUCCESS) << ",uid:" << effective_uid() 
           << ",username:\"" << base::JsonEscape(username_) << "\",desc:\""
           << (is_access_switch ? "switch ok" : "login ok.") << "\"}";

  if (!callback.empty()) {
    rsp_body << ")";
  }

  std::stringstream rsp_header;
  rsp_header << "HTTP/1.1 200 OK\r\n"
             << "Content-Type: text/javascript; charset=utf-8\r\n"
             << "Connection: close\r\n"
             << "Content-Length: " << rsp_body.str().size() << "\r\n"
             << IdCookiesHeader(qid_, qid_ticket_, temp_id_, temp_id_ticket_)
             << "\r\n";

  response_ = rsp_header.str();
  response_.append(rsp_body.str());
  connection_->WriteResponse(response_);

  UidSessionMap::Instance().Add(effective_uid(), effective_session());
  GetThreadManager()->add(boost::shared_ptr<Runnable>(new UpdateSessionTask(effective_uid(), effective_session(), true)));

  GetThreadManager()->add(boost::shared_ptr<Runnable>(new NotifyPresenceTask(effective_uid(), effective_session(), SessionType::WEB)));

  if (qid_ > 0 && temp_id_ > 0) {
    GetThreadManager()->add(boost::shared_ptr<Runnable>(new LinkTempIdQidTask(temp_id_, qid_)));
    GetThreadManager()->add(boost::shared_ptr<Runnable>(new UpdateSessionTask(temp_id_, temp_id_ticket_, false)));
  }
  return 0;
}

int CometSubscribeRequest::HostSwitchResponse() {
  // 用户在多个接入服务之间切换的情况
  // not implemented
  return 0;
}
int CometSubscribeRequest::IeBugFixResponse() {
  // TODO : 分配登录或临时身份
  std::stringstream rsp_body;

  rsp_body << "HTTP/1.1 200 OK\r\n"
           << "Content-Type: text/javascript; charset=utf-8\r\n"
           << "Connection: close\r\n"
           << "Content-Length: 1\r\n"
           << "\r\n"
           << '{';
  connection_->WriteResponse(rsp_body.str());
  return 0;
}

int CometSubscribeRequest::DuplicateErrorResponse() {
  // TODO : 分配登录或临时身份
  std::stringstream rsp_body;
  std::string callback = ParseCallback();
  if (!callback.empty()) {
    // 开头空格是 IE bug fix
    rsp_body << ' ' << callback << "(";
  }
  rsp_body << "{type:" << DUPLICATE_SUBSCRIBER 
           << ",desc:\"dup_conn\"}";

  if (!callback.empty()) {
    rsp_body << ")";
  }

  std::stringstream rsp_header;
  rsp_header << "HTTP/1.1 400 Bad Request\r\n"
             << "Content-Type: text/javascript; charset=utf-8\r\n"
             << "Connection: close\r\n"
             << "Content-Length: " << rsp_body.str().size() << "\r\n"
             << "\r\n";

  response_ = rsp_header.str();
  response_.append(rsp_body.str());
  connection_->WriteResponse(response_);

  LOG_WARN("Duplicate subscriber=" << effective_uid() << " sesssion=" << effective_session() << " addr=" << remote_addr());
  return 0;
}


void CometSubscribeRequest::AsyncWaitForMessage() {
  ImMessageList * buffered_messages = LingerSessionMap::Instance().Revive(effective_session());
  if (SubscriberCache::Instance().Add(effective_session(), shared_from_this())) {
    LOG_INFO("WaitForMessage uid=" << effective_uid() << " sesssion=" << effective_session() << " addr=" << remote_addr());
  } else {
    DuplicateErrorResponse();
    return;
  }

  if (buffered_messages) {
    OnMessageReady(*buffered_messages);
    delete buffered_messages;
    buffered_messages = NULL;
  }

  UidSessionMap::Instance().Add(effective_uid(), effective_session());
  GetThreadManager()->add(boost::shared_ptr<Runnable>(new UpdateSessionTask(effective_uid(), effective_session(), true)));
}

bool CometSubscribeRequest::Response() {
  if (!query_param("rx").empty()) {
    IeBugFixResponse();
    return true;
  }

  if (CheckTicket() != 0) {
    // 从Session/UserProfile取认证信息
    GetThreadManager()->add(boost::shared_ptr<Runnable>(new CheckTicketTask(shared_from_this(), qid_, qid_ticket_, cookie("Q"), cookie("T"))));
    return true;
  }

  if (effective_uid() <= 0) {
    temp_id_ = TempIdGenerator::Instance().Generate();
    temp_id_ticket_ = CalculateSession(temp_id_, "", "");
    LOG_INFO("brand new user " << temp_id_ << " login, session=" << temp_id_ticket_);
    UserLoginResponse(false);
  } else if (UidSessionMap::Instance().Contains(effective_uid(), effective_session())) {
    AsyncWaitForMessage();
  } else {
    UserLoginResponse(true); // access move
  }

  return true;
}

void CometSubscribeRequest::OnWriteError() {
  LOG_INFO("CometSubscribeRequest write error : " << effective_uid() << " " << effective_session());
  AddLingerSession();
}


void CometSubscribeRequest::OnWriteComplete() {
  LOG_INFO("CometSubscribeRequest write complete : " << effective_uid() << " " << effective_session());
  AddLingerSession();
}

int CometSubscribeRequest::OnMessageReady(const ImMessageList & messages) {
  ++out_messages_;
  std::stringstream rsp_body;

  std::string callback = ParseCallback();
  if (!callback.empty()) {
    // 开头空格是 IE bug fix
    rsp_body << ' ' << callback << "(";
  }
  rsp_body << '[';
  for(size_t i = 0; i < messages.size(); ++i) {
    if (i > 0) {
      rsp_body << ',';
    }
    LOG_INFO("OnMessageReady sender=" << messages[i].sender << " receiver=" << messages[i].receiver
              << " message=" << messages[i].body << " addr=" << remote_addr());
    rsp_body << "{type:" << IM_MESSAGE
       << ",subtype:" << MessageType(messages[i])
       << ",sender:" << messages[i].sender 
       << ",receiver:" << messages[i].receiver
       << ",timestamp:" << messages[i].timestamp
       << ",message:\"" << base::JsonEscape(messages[i].body);

    if (messages[i].__isset.via_shop) {
      rsp_body << "\", \"via_shop\":\"" << messages[i].via_shop;
      if (messages[i].__isset.via_shop_name) {
        rsp_body << "\", \"via_shop_name\":\"" << base::JsonEscape(messages[i].via_shop_name);
      }
      if (messages[i].__isset.via_shop_url) {
        rsp_body << "\", \"via_shop_url\":\"" << messages[i].via_shop_url;
      }
    }

    rsp_body << "\"}";
  }
  rsp_body << ']';
  if (!callback.empty()) {
    rsp_body << ")";
  }

  response_ = "HTTP/1.1 200 OK\r\nContent-Type: text/javascript; charset=utf-8\r\nConnection: close\r\nContent-Length: ";
  response_ += boost::lexical_cast<std::string>(rsp_body.str().size());
  response_.append("\r\n\r\n");
  response_.append(rsp_body.str());
  connection_->WriteResponse(response_);

  return 0;
}

void CometSubscribeRequest::AddLingerSession() {
  SubscriberCache::Instance().Remove(effective_session(), shared_from_this());
  LingerSessionMap::Instance().Add(effective_uid(), effective_session());
}

}


